import asyncio
import json
import logging
import subprocess
import os
import socket
import uuid
import websockets
import concurrent.futures
import struct
from typing import Dict, Optional
from dataclasses import dataclass
from enum import Enum
from threading import Event
import urllib.parse
import time
import aiohttp
import redis
import base64
from Crypto.Cipher import AES

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler('/app/logs/ffmpeg_service.log')
    ]
)
logger = logging.getLogger(__name__)

# 本地缓存目录
CACHE_DIR = '/app/cache'
os.makedirs(CACHE_DIR, exist_ok=True)

# XOR加密密钥
XOR_KEY = b'falconix2024'

def decrypt_url(encrypted_url):
    try:
        # 先进行Base64解码
        encrypted_data = base64.urlsafe_b64decode(encrypted_url)
        # 创建解密后的字节数组
        decrypted_data = bytearray(len(encrypted_data))
        # 执行XOR解密
        for i in range(len(encrypted_data)):
            decrypted_data[i] = encrypted_data[i] ^ XOR_KEY[i % len(XOR_KEY)]
        # 返回解密后的URL
        return decrypted_data.decode()
    except Exception as e:
        logger.error(f"URL解密失败: {str(e)}")
        return encrypted_url

def get_local_file_path(url: str) -> str:
    """从URL中提取文件名并返回本地文件路径"""
    try:
        parsed_url = urllib.parse.urlparse(url)
        filename = os.path.basename(parsed_url.path)
        if not filename:
            raise ValueError("URL中未找到文件名")
        return os.path.join(CACHE_DIR, filename)
    except Exception as e:
        logger.error(f"解析URL失败: {str(e)}")
        return None

async def download_file(url: str, local_path: str) -> bool:
    """从URL下载文件到本地"""
    try:
        logger.info(f"开始下载文件: {url} -> {local_path}")
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                if response.status == 200:
                    # 使用aiohttp的stream方式下载文件
                    with open(local_path, 'wb') as f:
                        async for chunk in response.content.iter_chunked(1024*1024):  # 1MB chunks
                            f.write(chunk)
                    logger.info(f"文件下载完成: {local_path}")
                    return True
                else:
                    logger.error(f"下载失败，HTTP状态码: {response.status}")
                    return False
    except Exception as e:
        logger.error(f"下载文件时出错: {str(e)}")
        return False

async def ensure_local_file(url: str) -> Optional[str]:
    """确保文件在本地存在，如果不存在则下载"""
    local_path = get_local_file_path(url)
    if not local_path:
        return None
        
    if os.path.exists(local_path):
        logger.info(f"使用本地缓存文件: {local_path}")
        return local_path
        
    if await download_file(url, local_path):
        return local_path
    return None

class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    ERROR = "error"

@dataclass
class StreamTask:
    process: asyncio.subprocess.Process
    status: TaskStatus
    websocket: websockets.WebSocketServerProtocol
    stop_event: Event
    task_id: str
    url: str
    seek_time: str = None
    watermark_text: str = None
    tcp_server: socket.socket = None
    output_thread: asyncio.Task = None
    error_thread: asyncio.Task = None
    client_socket: socket.socket = None
    input_stream: socket.socket = None
    ffmpeg_output_lock = asyncio.Lock()  # 添加锁

    def __init__(self, process, status, websocket, stop_event, task_id, url, seek_time=None, watermark_text=None, tcp_server=None):
        self.process = process
        self.status = status
        self.websocket = websocket
        self.stop_event = stop_event
        self.task_id = task_id
        self.url = url
        self.seek_time = seek_time
        self.watermark_text = watermark_text
        self.tcp_server = tcp_server
        self.output_thread = None
        self.error_thread = None
        self.client_socket = None
        self.input_stream = None
        self.ffmpeg_output_lock = asyncio.Lock()  # 添加锁

    async def send_data(self, data):
        """向客户端发送数据"""
        try:
            await self.websocket.send(data)
        except Exception as e:
            logger.error(f"发送数据失败: 任务ID={self.task_id}, 错误={str(e)}")
            self.stop_event.set()

    async def cleanup(self):
        """清理资源"""
        if self.process and self.process.returncode is None:
            try:
                logger.info(f"正在终止FFmpeg进程: pid={self.process.pid}")
                self.process.terminate()
                try:
                    await asyncio.wait_for(self.process.wait(), timeout=2)
                    logger.info(f"FFmpeg进程已终止: pid={self.process.pid}")
                except asyncio.TimeoutError:
                    logger.warning(f"FFmpeg进程终止超时，强制结束: pid={self.process.pid}")
                    self.process.kill()
            except Exception as e:
                logger.error(f"终止FFmpeg进程时出错: pid={self.process.pid}, 错误={str(e)}")
        
        if self.tcp_server:
            try:
                logger.info(f"正在关闭TCP服务器: {self.tcp_server.getsockname()}")
                self.tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
                self.tcp_server.shutdown(socket.SHUT_RDWR)
                self.tcp_server.close()
                logger.info("TCP服务器已关闭")
            except Exception as e:
                logger.error(f"关闭TCP服务器时出错: {str(e)}")
        
        if self.output_thread and not self.output_thread.done():
            logger.info("取消输出线程")
            self.output_thread.cancel()
        
        if self.error_thread and not self.error_thread.done():
            logger.info("取消错误监控线程")
            self.error_thread.cancel()

@dataclass
class TranscodeTask:
    process: asyncio.subprocess.Process
    status: TaskStatus
    websocket: websockets.WebSocketServerProtocol
    stop_event: Event
    task_id: str
    input_path: str
    output_path: str
    progress_thread: asyncio.Task = None
    error_thread: asyncio.Task = None

    async def send_progress(self, progress):
        """向客户端发送进度"""
        try:
            await self.websocket.send(json.dumps({
                'status': 'progress',
                'task_id': self.task_id,
                'progress': progress
            }))
        except Exception as e:
            logger.error(f"发送进度失败: 任务ID={self.task_id}, 错误={str(e)}")
            self.stop_event.set()

    async def cleanup(self):
        """清理资源"""
        if self.process and self.process.returncode is None:
            try:
                self.process.terminate()
                try:
                    await asyncio.wait_for(self.process.wait(), timeout=2)
                except asyncio.TimeoutError:
                    self.process.kill()
                logger.info(f"FFmpeg进程已终止: 任务ID={self.task_id}")
            except Exception as e:
                logger.error(f"终止FFmpeg进程时出错: 任务ID={self.task_id}, 错误={str(e)}")
        
        if self.progress_thread and not self.progress_thread.done():
            self.progress_thread.cancel()
        
        if self.error_thread and not self.error_thread.done():
            self.error_thread.cancel()

class FFmpegService:
    def __init__(self, host='0.0.0.0', port=15000):
        self.host = host
        self.port = port
        self.stream_tasks = {}  # 存储所有流任务，key为task_id
        self.tcp_servers = {}  # 存储TCP服务器，key为task_id
        self.transcode_tasks = {}  # 存储所有转码任务
        self.process_executor = concurrent.futures.ThreadPoolExecutor(max_workers=100)
        self.server = None
        self.loop = None
        self.used_ports = set()  # 记录已使用的端口
        self.port_lock = asyncio.Lock()  # 端口分配锁

        # Redis连接配置
        self.redis_host = '172.29.39.249'  # Redis服务器地址
        self.redis_port = 11852        # Redis端口
        self.redis_db = 0            # 数据库编号
        self.redis_password = '21XGsc1m6cI5mVsy'  # Redis密码
        self.redis_client = None      # Redis客户端实例

        logger.info(f"初始化FFmpeg服务: host={host}, port={port}")

        def init_redis(self):
            """初始化Redis连接"""
            try:
                self.redis_client = redis.Redis(
                    host=self.redis_host,
                    port=self.redis_port,
                    db=self.redis_db,
                    password=self.redis_password,  # 添加密码
                    decode_responses=True,
                    socket_timeout=5,  # 设置超时时间
                    socket_connect_timeout=5
                )
                # 测试连接
                self.redis_client.ping()
                logger.info("Redis连接成功")
            except redis.ConnectionError as e:
                logger.error(f"Redis连接失败: {str(e)}")
                raise
            except Exception as e:
                logger.error(f"初始化Redis时出错: {str(e)}")
                raise

    async def get_user_info(self, token: str) -> Optional[dict]:
        """从Redis获取用户信息"""
        if not token:
            return None

        try:
            # 确保Redis连接已初始化
            if not self.redis_client:
                try:
                    self.redis_client = redis.Redis(
                        host=self.redis_host,
                        port=self.redis_port,
                        db=self.redis_db,
                        password=self.redis_password,  # 添加密码
                        decode_responses=True,
                        socket_timeout=5,  # 设置超时时间
                        socket_connect_timeout=5
                    )
                    # 测试连接
                    self.redis_client.ping()
                    logger.info("Redis连接成功")
                except redis.ConnectionError as e:
                    logger.error(f"Redis连接失败: {str(e)}")
                    return None
                except Exception as e:
                    logger.error(f"初始化Redis时出错: {str(e)}")
                    return None

            redis_key = f"shiro_session_key:{token}"
            user_info = self.redis_client.get(redis_key)

            if not user_info:
                logger.warning(f"未找到用户信息: token={token}")
                return None

            # 解析JSON数据
            user_data = json.loads(user_info)
            if isinstance(user_data, list) and len(user_data) == 2:
                user_info = user_data[1]
                if isinstance(user_info, dict):
                    return user_info
            return None

        except json.JSONDecodeError as e:
            logger.error(f"解析用户信息JSON失败: {str(e)}")
            return None
        except Exception as e:
            logger.error(f"获取用户信息时出错: {str(e)}")
            return None

    async def get_available_port(self, start_port=15001, end_port=16000):
        """获取可用端口"""
        async with self.port_lock:
            for port in range(start_port, end_port):
                if port not in self.used_ports:
                    try:
                        # 尝试绑定端口
                        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                        sock.bind(('127.0.0.1', port))
                        sock.close()
                        self.used_ports.add(port)
                        return port
                    except socket.error:
                        continue
            raise Exception("无法找到可用端口")

    async def release_port(self, port):
        """释放端口"""
        async with self.port_lock:
            if port in self.used_ports:
                self.used_ports.remove(port)

    async def get_tcp_output(self, task_id):
        """创建TCP服务器并返回输出地址"""
        try:
            logger.info(f"开始创建TCP服务器: task_id={task_id}")
            tcp_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
            
            # 增加接收和发送缓冲区大小
            tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 32*1024*1024)  # 32MB
            tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 32*1024*1024)  # 32MB
            
            # 设置TCP_NODELAY，禁用Nagle算法
            tcp_server.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            
            # 设置TCP_KEEPALIVE
            tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
            tcp_server.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 60)  # 空闲60秒后开始探测
            tcp_server.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10)  # 每10秒探测一次
            tcp_server.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3)     # 最多探测3次
            
            # 获取可用端口
            port = await self.get_available_port()
            tcp_server.bind(('127.0.0.1', port))
            logger.info(f"TCP服务器绑定端口成功: port={port}")
            
            tcp_server.listen(1)
            tcp_server.setblocking(False)  # 设置为非阻塞模式
            
            host, port = tcp_server.getsockname()
            output = f"tcp://{host}:{port}"
            
            self.tcp_servers[task_id] = tcp_server
            logger.info(f"TCP服务器创建成功: task_id={task_id}, address={output}")
            return output, tcp_server
            
        except Exception as e:
            logger.error(f"创建TCP服务器失败: task_id={task_id}, 错误={str(e)}", exc_info=True)
            raise

    async def check_ffmpeg_status(self, task: StreamTask):
        """检查FFmpeg进程状态"""
        try:
            # 检查进程是否还在运行
            if task.process.returncode is not None:
                logger.error(f"FFmpeg进程已退出: 任务ID={task.task_id}, 返回码={task.process.returncode}")
                return False
            
            # 检查进程是否卡住
            try:
                async with task.ffmpeg_output_lock:
                    # 尝试读取一行错误输出，设置超时
                    line = await asyncio.wait_for(task.process.stderr.readline(), timeout=0.1)
                    if line:
                        line = line.decode().strip()
                        # 检查是否为进度信息
                        if any(x in line for x in ['frame=', 'fps=', 'q=', 'size=', 'time=', 'bitrate=', 'speed=']):
                            logger.debug(f"FFmpeg进度: 任务ID={task.task_id}, 进度={line}")
            except Exception:
                # 忽略读取错误，只要进程还在运行就继续
                pass
            
            return True
        except Exception as e:
            logger.error(f"检查FFmpeg状态时出错: 任务ID={task.task_id}, 错误={str(e)}")
            return True  # 返回True，不终止进程

    async def check_file_status(self, file_path: str) -> bool:
        """检查文件状态"""
        try:
            if not os.path.exists(file_path):
                logger.error(f"文件不存在: {file_path}")
                return False
            
            if not os.access(file_path, os.R_OK):
                logger.error(f"文件不可读: {file_path}")
                return False
            
            # 检查文件是否正在被写入
            try:
                with open(file_path, 'r') as f:
                    f.seek(0, 2)  # 移动到文件末尾
                    size1 = f.tell()
                    await asyncio.sleep(1)
                    f.seek(0, 2)
                    size2 = f.tell()
                    if size1 != size2:
                        logger.warning(f"文件正在被写入: {file_path}")
            except Exception as e:
                logger.error(f"检查文件状态时出错: {file_path}, 错误={str(e)}")
            
            return True
        except Exception as e:
            logger.error(f"检查文件状态时出错: {file_path}, 错误={str(e)}")
            return False

    async def forward_stream_data(self, task: StreamTask):
        """转发流数据"""
        logger.info(f"开始转发流数据: 任务ID={task.task_id}")
        
        try:
            # 等待TCP客户端连接
            logger.info(f"等待TCP客户端连接: 任务ID={task.task_id}")
            task.client_socket, _ = await self.loop.sock_accept(task.tcp_server)
            logger.info(f"TCP客户端已连接: 任务ID={task.task_id}")
            
            # 初始化变量
            buffer = bytearray()
            total_bytes = 0
            last_data_time = time.time()
            no_data_count = 0
            chunk_size = 1024
            header_sent = False
            metadata_sent = False
            process_ended = False
            last_log_time = time.time()
            start_time = time.time()
            buffer_size_limit = 32*1024*1024
            last_buffer_check = time.time()
            buffer_check_interval = 1.0
            last_status_check = time.time()
            status_check_interval = 5.0
            ws_buffer_size = 0
            last_transfer_log_time = time.time()
            transfer_log_interval = 10.0
            last_ffmpeg_check = time.time()
            ffmpeg_check_interval = 5.0

            while not task.stop_event.is_set():
                try:
                    current_time = time.time()
                    
                    # 定期检查FFmpeg状态
                    if current_time - last_ffmpeg_check >= ffmpeg_check_interval:
                        await self.check_ffmpeg_status(task)  # 只检查状态，不终止进程
                        last_ffmpeg_check = current_time

                    # 检查WebSocket连接状态
                    if task.websocket.closed:
                        logger.info(f"检测到WebSocket连接已关闭: 任务ID={task.task_id}")
                        await self.cleanup_task(task.task_id)
                        return

                    # 从TCP连接读取数据
                    try:
                        data = await self.loop.sock_recv(task.client_socket, chunk_size)
                        if not data:
                            no_data_count += 1
                            if no_data_count > 5:
                                if buffer:
                                    logger.info(f"发送缓冲区中剩余的数据: 任务ID={task.task_id}, 大小={len(buffer)}")
                                    if not task.websocket.closed:
                                        await task.send_data(bytes(buffer))
                                        buffer.clear()
                                logger.warning(f"连续{no_data_count}次没有数据可读: 任务ID={task.task_id}")
                                break
                            await asyncio.sleep(0.1)
                            continue

                        # 重置无数据计数
                        no_data_count = 0
                        last_data_time = time.time()

                        # 将数据添加到缓冲区
                        buffer.extend(data)
                        total_bytes += len(data)

                        # 处理FLV头部
                        if not header_sent and len(buffer) >= 9:
                            if not task.websocket.closed:
                                await task.send_data(bytes(buffer[:9]))
                            header_sent = True
                            logger.info(f"已发送FLV头部: 任务ID={task.task_id}, 长度=9")
                            await asyncio.sleep(0.1)
                            buffer = buffer[9:]
                            continue

                        # 处理元数据
                        if not metadata_sent and len(buffer) >= 100:
                            if not task.websocket.closed:
                                await task.send_data(bytes(buffer))
                            metadata_sent = True
                            logger.info(f"已发送元数据: 任务ID={task.task_id}, 长度={len(buffer)}")
                            buffer.clear()
                            continue

                        # 处理帧数据
                        if len(buffer) > 0:
                            try:
                                if not task.websocket.closed:
                                    await task.send_data(bytes(buffer))
                                    buffer.clear()
                            except Exception as e:
                                logger.error(f"发送数据失败: 任务ID={task.task_id}, 错误={str(e)}")
                                await asyncio.sleep(0.1)

                        # 定期记录传输日志
                        if current_time - last_transfer_log_time >= transfer_log_interval:
                            logger.info(f"数据传输统计: 任务ID={task.task_id}, 已传输={total_bytes/1024/1024:.2f}MB, 耗时={(current_time - start_time):.2f}秒, 缓冲区大小={len(buffer)/1024/1024:.2f}MB")
                            last_transfer_log_time = current_time

                    except BlockingIOError:
                        await asyncio.sleep(0.1)
                        continue
                    except Exception as e:
                        logger.error(f"读取数据出错: 任务ID={task.task_id}, 错误={str(e)}")
                        break

                except Exception as e:
                    logger.error(f"处理数据出错: 任务ID={task.task_id}, 错误={str(e)}")
                    break

            # 确保所有数据都已发送
            if buffer:
                logger.info(f"发送最后的数据包: 任务ID={task.task_id}, 大小={len(buffer)}")
                if not task.websocket.closed:
                    await task.send_data(bytes(buffer))
                    buffer.clear()

            logger.info(f"数据传输完成: 任务ID={task.task_id}, 总耗时={time.time() - start_time:.2f}秒, 总字节数={total_bytes/1024/1024:.2f}MB")

        except Exception as e:
            error_msg = f"转发数据出错: {str(e)}"
            logger.error(f"{error_msg}, 任务ID={task.task_id}")
            if not task.websocket.closed:
                try:
                    await task.websocket.send(json.dumps({
                        'status': 'error',
                        'task_id': task.task_id,
                        'error': error_msg
                    }))
                except:
                    pass
        finally:
            # 只在WebSocket连接关闭时才清理资源
            if task.websocket.closed:
                # 立即清理资源
                if task.client_socket:
                    try:
                        task.client_socket.shutdown(socket.SHUT_RDWR)
                        task.client_socket.close()
                        logger.info(f"已关闭客户端socket: 任务ID={task.task_id}")
                    except Exception as e:
                        logger.error(f"关闭客户端socket出错: 任务ID={task.task_id}, 错误={str(e)}")
                await self.cleanup_task(task.task_id)

    async def handle_websocket(self, websocket, path):
        """处理WebSocket连接"""
        task_id = None
        try:
            logger.info(f"收到新的WebSocket连接请求: path={path}")

            # 记录原始请求信息
            logger.info("========================= 新的WebSocket连接 =========================")
            logger.info(f"请求路径: {path}")
            logger.info("请求头信息:")
            for header, value in websocket.request_headers.items():
                logger.info(f"  {header}: {value}")
            logger.info(f"远程地址: {websocket.remote_address}")
            logger.info("============================================================")

            # 处理带有pywss前缀的路径
            original_path = path
            if path.startswith('/pywss'):
                path = path.replace('/pywss', '', 1)  # 只替换第一个匹配项
                logger.info(f"- 移除pywss前缀后的路径: {path}")

            # 处理带有pywss前缀的路径
            if path.startswith('/pywss'):
                path = path.replace('/pywss', '', 1)  # 只替换第一个匹配项
                logger.info(f"移除pywss前缀后的路径: {path}")

            params = {}
            if '?' in path:
                query = path.split('?')[1]
                for param in query.split('&'):
                    # 只分割第一个等号
                    parts = param.split('=', 1)
                    if len(parts) == 2:
                        key, value = parts
                        params[key] = value
                    else:
                        logger.warning(f"忽略无效参数: {param}")

            logger.info(f"解析后的参数: params={params}")

            # 验证token
            token = params.get('token')
            if not token:
                error_msg = "缺少token参数"
                logger.warning(error_msg)
                await websocket.close(code=1008, reason=error_msg)
                return

            # 获取用户信息
            user_info = await self.get_user_info(token)
            if not user_info:
                error_msg = "无效的token或用户信息不存在"
                logger.warning(error_msg)
                await websocket.close(code=1008, reason=error_msg)
                return

            if 'type' not in params:
                logger.warning("缺少type参数")
                await websocket.close(code=1008, reason='Missing type parameter')
                return

            if params['type'] == 'stream':
                logger.info("开始处理流请求")
                task_id = str(uuid.uuid4())  # 生成唯一的task_id
                try:
                    await self.handle_stream_request(websocket, params, task_id, user_info)
                except websockets.exceptions.ConnectionClosed:
                    logger.info(f"WebSocket连接已关闭: task_id={task_id}")
                    await self.cleanup_task(task_id)
            elif params['type'] == 'transcode':
                await self.handle_transcode_request(websocket, params)
            else:
                logger.warning(f"无效的type参数: {params['type']}")
                await websocket.close(code=1008, reason='Invalid type parameter')

        except Exception as e:
            logger.error(f"处理WebSocket连接时出错: {str(e)}", exc_info=True)
            if not websocket.closed:
                await websocket.close(code=1011, reason=str(e))
            if task_id:
                await self.cleanup_task(task_id)

    async def cleanup_task(self, task_id):
        """清理指定task_id的所有资源"""
        if task_id not in self.stream_tasks:
            logger.warning(f"任务不存在，无需清理: task_id={task_id}")
            return

        task = self.stream_tasks[task_id]
        if not task:
            logger.warning(f"任务对象为空，无需清理: task_id={task_id}")
            return

        logger.info(f"开始清理任务资源: task_id={task_id}")

        try:
            # 设置停止标志
            task.stop_event.set()

            # 关闭TCP客户端连接
            if task.client_socket:
                try:
                    task.client_socket.shutdown(socket.SHUT_RDWR)
                    task.client_socket.close()
                    logger.info(f"TCP客户端连接已关闭: task_id={task_id}")
                except Exception as e:
                    logger.error(f"关闭TCP客户端连接时出错: task_id={task_id}, 错误={str(e)}")

            # 关闭TCP服务器
            if task.tcp_server:
                try:
                    task.tcp_server.shutdown(socket.SHUT_RDWR)
                    task.tcp_server.close()
                    logger.info(f"TCP服务器已关闭: task_id={task_id}")
                except Exception as e:
                    logger.error(f"关闭TCP服务器时出错: task_id={task_id}, 错误={str(e)}")

            # 停止FFmpeg进程
            if task.process and task.process.returncode is None:
                try:
                    logger.info(f"正在停止FFmpeg进程: task_id={task_id}, PID={task.process.pid}")
                    task.process.terminate()
                    try:
                        await asyncio.wait_for(task.process.wait(), timeout=2)
                        logger.info(f"FFmpeg进程已终止: task_id={task_id}, PID={task.process.pid}")
                    except asyncio.TimeoutError:
                        logger.warning(f"FFmpeg进程终止超时，强制结束: task_id={task_id}, PID={task.process.pid}")
                        task.process.kill()
                        try:
                            await asyncio.wait_for(task.process.wait(), timeout=1)
                            logger.info(f"FFmpeg进程已强制终止: task_id={task_id}, PID={task.process.pid}")
                        except asyncio.TimeoutError:
                            logger.error(f"无法终止FFmpeg进程: task_id={task_id}, PID={task.process.pid}")
                except Exception as e:
                    logger.error(f"终止FFmpeg进程时出错: task_id={task_id}, 错误={str(e)}")

            # 从任务列表中移除
            self.stream_tasks.pop(task_id, None)
            self.tcp_servers.pop(task_id, None)

            # 释放端口
            if task.tcp_server:
                try:
                    _, port = task.tcp_server.getsockname()
                    await self.release_port(port)
                    logger.info(f"已释放端口: {port}")
                except Exception as e:
                    logger.error(f"释放端口时出错: task_id={task_id}, 错误={str(e)}")

            logger.info(f"任务资源清理完成: task_id={task_id}")

        except Exception as e:
            logger.error(f"清理任务资源时出错: task_id={task_id}, 错误={str(e)}")
            raise

    async def handle_stream_request(self, ws, params, task_id, user_info):
        """处理流请求"""
        encrypted_url = params.get('url')
        if not encrypted_url:
            logger.error("URL 不能为空")
            await ws.close(code=1008, reason='URL is required')
            return

        # 解密URL
        url = decrypt_url(encrypted_url)
        logger.info(f"解密后的URL: {url}")

        # 从用户信息中获取水印文本
        watermark_text = f"{user_info.get('nickName', '')}{user_info.get('userId', '')}"
        if not watermark_text.strip(':'):  # 如果获取失败，使用默认值
            watermark_text = "鹰眼智能"

        logger.info(f"开始处理流请求: task_id={task_id}, url={url}")

        try:
            if not url:
                logger.error("URL 不能为空")
                await ws.close(code=1008, reason='URL is required')
                return

            # 确保文件在本地存在
            local_path = await ensure_local_file(url)
            if not local_path:
                error_msg = "无法获取视频文件"
                logger.error(f"{error_msg}, task_id={task_id}")
                await ws.send(json.dumps({
                    'status': 'error',
                    'error': error_msg
                }))
                return

            # 先创建TCP服务器
            tcp_output, tcp_server = await self.get_tcp_output(task_id)
            
            # 创建任务对象
            stop_event = Event()
            task = StreamTask(
                process=None,
                status=TaskStatus.PENDING,
                websocket=ws,
                stop_event=stop_event,
                task_id=task_id,
                url=local_path,
                seek_time=params.get('seek_time'),
                watermark_text=watermark_text,
                tcp_server=tcp_server
            )
            self.stream_tasks[task_id] = task
            self.tcp_servers[task_id] = tcp_server

            # 构建并启动FFmpeg进程
            cmd = self.build_ffmpeg_command(local_path, tcp_output, params.get('seek_time'), watermark_text)
            process = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                preexec_fn=os.setsid
            )

            logger.info(f"FFmpeg进程创建成功: task_id={task_id}, pid={process.pid}")
            task.process = process
            task.status = TaskStatus.RUNNING

            # 发送任务开始响应
            await ws.send(json.dumps({
                'status': 'started',
                'task_id': task_id
            }))

            # 启动数据转发任务
            task.output_thread = asyncio.create_task(self.forward_stream_data(task))

            # 启动错误监控
            task.error_thread = asyncio.create_task(self.monitor_ffmpeg_errors(task))

            # 等待任务完成
            await asyncio.gather(
                task.output_thread,
                task.error_thread,
                return_exceptions=True
            )

        except Exception as e:
            error_msg = f"推流任务出错: {str(e)}"
            logger.error(f"{error_msg}, task_id={task_id}")
            await ws.send(json.dumps({
                'status': 'error',
                'error': error_msg
            }))
            raise
        finally:
            await self.cleanup_task(task_id)

    async def handle_transcode_request(self, websocket, params):
        """处理转码请求"""
        task_id = params.get('task_id', str(uuid.uuid4()))
        input_path = params.get('input_path')
        output_path = params.get('output_path')

        if not all([input_path, output_path]):
            await websocket.close(code=1008, reason='Missing required parameters')
            return

        try:
            # 构建ffmpeg命令
            cmd = [
                'ffmpeg',
                '-i', input_path,
                '-c:v', 'libx264',
                '-preset', 'medium',
                '-crf', '23',
                '-c:a', 'aac',
                '-b:a', '128k',
                '-y',  # 覆盖输出文件
                output_path
            ]

            logger.info(f"执行FFmpeg命令: {' '.join(cmd)}, 任务ID={task_id}")

            # 启动FFmpeg进程
            process = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                universal_newlines=True
            )

            # 创建任务对象
            stop_event = Event()
            task = TranscodeTask(
                process=process,
                status=TaskStatus.RUNNING,
                websocket=websocket,
                stop_event=stop_event,
                task_id=task_id,
                input_path=input_path,
                output_path=output_path
            )

            # 添加任务
            self.transcode_tasks[task_id] = task

            # 发送任务开始响应
            await websocket.send(json.dumps({
                'status': 'started',
                'task_id': task_id
            }))

            # 启动进度监控和错误监控
            task.progress_thread = asyncio.create_task(self.monitor_transcode_progress(task))
            task.error_thread = asyncio.create_task(self.monitor_ffmpeg_errors(task))

            # 等待任务完成
            await asyncio.gather(
                task.progress_thread,
                task.error_thread
            )

        except Exception as e:
            error_msg = f"转码任务出错: {str(e)}"
            logger.error(f"{error_msg}, 任务ID={task_id}")
            await websocket.send(json.dumps({
                'status': 'error',
                'error': error_msg
            }))
            raise
        finally:
            # 清理资源
            if task_id in self.transcode_tasks:
                task = self.transcode_tasks.pop(task_id)
                if task:
                    await task.cleanup()

    async def monitor_ffmpeg_errors(self, task: StreamTask):
        """监控FFmpeg错误输出"""
        try:
            while not task.stop_event.is_set():
                if task.process.returncode is not None:
                    async with task.ffmpeg_output_lock:  # 使用锁
                        error_output = await task.process.stderr.read()
                        if error_output:
                            error_text = error_output.decode()
                            if "ffmpeg version" not in error_text:
                                logger.info(f"FFmpeg进程结束输出: 任务ID={task.task_id}, 输出={error_text}")
                    break

                try:
                    async with task.ffmpeg_output_lock:  # 使用锁
                        error_line = await asyncio.wait_for(task.process.stderr.readline(), timeout=0.1)
                        if error_line:
                            error_text = error_line.decode().strip()
                            # 忽略版本信息和配置信息
                            if any(x in error_text for x in ['ffmpeg version', 'configuration:', 'libav', 'built with']):
                                continue
                            
                            # 检查是否为流信息
                            if any(x in error_text for x in ['Stream #', 'Metadata:', 'handler_name', 'vendor_id', 'encoder', 'cpb:', 'vbv_delay']):
                                logger.info(f"FFmpeg流信息: 任务ID={task.task_id}, 信息={error_text}")
                                continue
                            
                            # 检查是否为进度信息
                            if any(x in error_text for x in ['frame=', 'fps=', 'q=', 'size=', 'time=', 'bitrate=', 'speed=']):
                                logger.debug(f"FFmpeg进度: 任务ID={task.task_id}, 进度={error_text}")
                                continue
                            
                            # 检查是否为可容忍的错误
                            if any(x in error_text for x in [
                                'partial file',
                                'corrupt input packet',
                                'Input buffer exhausted',
                                'Invalid data found',
                                'decode_band_types',
                                'Separator is not found',
                                'chunk exceed the limit'
                            ]):
                                logger.debug(f"FFmpeg警告(可容忍): 任务ID={task.task_id}, 警告={error_text}")
                                continue
                            
                            # 其他错误信息
                            logger.error(f"FFmpeg错误: 任务ID={task.task_id}, 错误={error_text}")
                except asyncio.TimeoutError:
                    continue
                except Exception as e:
                    logger.error(f"读取FFmpeg输出时出错: 任务ID={task.task_id}, 错误={str(e)}")
                    break

                await asyncio.sleep(0.1)

        except Exception as e:
            logger.error(f"监控FFmpeg日志时出错: 任务ID={task.task_id}, 错误={str(e)}")

    async def monitor_transcode_progress(self, task: TranscodeTask):
        """监控转码进度"""
        try:
            while not task.stop_event.is_set():
                if task.process.returncode is not None:
                    # 进程已结束
                    if task.process.returncode == 0:
                        await task.websocket.send(json.dumps({
                            'status': 'completed',
                            'task_id': task.task_id
                        }))
                else:
                    await task.websocket.send(json.dumps({
                        'status': 'error',
                        'task_id': task.task_id,
                        'error': f"FFmpeg进程异常退出，返回码: {task.process.returncode}"
                    }))
                    break

                # 读取进度信息
                line = await task.process.stderr.readline()
                if line:
                    # 解析进度信息
                    if 'time=' in line:
                        time_str = line.split('time=')[1].split()[0]
                        await task.send_progress(time_str)

                await asyncio.sleep(0.1)

        except Exception as e:
            logger.error(f"监控转码进度时出错: 任务ID={task.task_id}, 错误={str(e)}")
            
    async def start(self):
        """启动服务"""
        try:
            self.loop = asyncio.get_event_loop()  # 获取当前事件循环
            logger.info(f"正在启动WebSocket服务器: host={self.host}, port={self.port}")
            self.server = await websockets.serve(
                self.handle_websocket,
                self.host,
                self.port,
                ping_interval=15,
                ping_timeout=10,
                max_size=100*1024*1024,
                max_queue=512,
                compression=None
            )
            logger.info(f"WebSocket服务器启动成功: {self.host}:{self.port}")
            await self.server.wait_closed()
        except Exception as e:
            logger.error(f"WebSocket服务器启动失败: {str(e)}", exc_info=True)
            raise
        finally:
            if self.server:
                self.server.close()
                await self.server.wait_closed()

    def build_ffmpeg_command(self, url, tcp_output, seek_time=None, watermark_text=None):
        """构建优化的FFmpeg命令"""
        cmd = [
            'ffmpeg',
            '-nostdin',  # 禁用标准输入防止阻塞[3,5](@ref)
            '-re',
            '-thread_queue_size', '1024',  # 提升输入缓冲[6,8](@ref)
            '-i', url,
            '-timeout', '3000000',  # 3秒输入流超时[1,2](@ref)
            '-reconnect', '1',  # 自动重连[5](@ref)
            '-reconnect_streamed', '1',
            '-reconnect_delay_max', '30'
        ]

        if seek_time:
            cmd.extend(['-ss', seek_time])

        if watermark_text:
            font_path = '/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc'
            drawtext = f"drawtext=fontfile={font_path}:text='{watermark_text}':fontcolor=gray@0.2:fontsize=42:x='min(max(50,w/2+(w/2-tw-100)*sin(2*PI*t/48)),w-tw-50)':y='min(max(50,h/2+(h/2-th-100)*cos(2*PI*t/32)),h-th-50)':shadowcolor=black@0.3:shadowx=2:shadowy=2"
            cmd.extend(['-vf', drawtext])

        cmd.extend([
            '-max_delay', '5000000',  # 增大延迟容忍[6](@ref)
            '-s', '1280x720',
            '-g', '30',  # 优化GOP大小[7,8](@ref)
            '-keyint_min', '30',  # 强制关键帧间隔[7](@ref)
            '-r', '30',
            '-c:v', 'libx264',
            '-b:v', '2000k',
            '-maxrate', '2000k',  # 匹配缓冲区比例[7](@ref)
            '-bufsize', '4000k',  # 2:1缓冲策略[7](@ref)
            '-preset', 'veryfast',  # 平衡速度与质量[6,8](@ref)
            '-tune', 'fastdecode',  # 提升解码兼容性[6](@ref)
            '-x264opts', 'no-mbtree',  # 禁用宏块树优化[7](@ref)
            '-c:a', 'aac',
            '-q:a', '4',  # 可变音频质量[3,6](@ref)
            '-ac', '2',
            '-ar', '44100',
            '-f', 'flv',
            '-vsync', 'cfr',
            '-async', '1',
            '-threads', '0',  # 自动分配线程[6,8](@ref)
            '-loglevel', 'verbose',  # 调试日志[3](@ref)
            '-y',  # 强制覆盖输出
            tcp_output
        ])
        return cmd



if __name__ == '__main__':
    service = FFmpegService()
    asyncio.run(service.start())